package com.icbc.dccsh.bolt;

import java.util.Map;

import org.apache.storm.redis.bolt.RedisLookupBolt;
import org.apache.storm.redis.common.config.JedisPoolConfig;
import org.apache.storm.redis.common.mapper.RedisLookupMapper;

import com.icbc.dccsh.util.Watcher;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;

public class RedisBoltBuilder {
	// private RedisLookupMapper _mapper = null;
	private RedisLookupBolt _bolt = null;
	private Watcher _watcher = null;

	public static RedisLookupBolt build(RedisLookupMapper mapper){
		String host = "localhost";
		int port = 6379;
		JedisPoolConfig poolConfig = new JedisPoolConfig.Builder().setHost(host).setPort(port).setDatabase(0).build();
		// RedisLookupMapper lookupMapper = new FormulaRedisLookupMapper();
		RedisLookupMapper lookupMapper = mapper;
		RedisLookupBolt lookupBolt = new RedisLookupBolt(poolConfig, lookupMapper);
		return lookupBolt;
//		this._bolt = lookupBolt;
	}
	private RedisBoltBuilder(){
		// Nothing to do.
	}
//	public RedisBoltBuilder(RedisLookupMapper mapper) {
//		String host = "localhost";
//		int port = 6379;
//		JedisPoolConfig poolConfig = new JedisPoolConfig.Builder().setHost(host).setPort(port).setDatabase(0).build();
//		// RedisLookupMapper lookupMapper = new FormulaRedisLookupMapper();
//		RedisLookupMapper lookupMapper = mapper;
//		RedisLookupBolt lookupBolt = new RedisLookupBolt(poolConfig, lookupMapper);
//		this._bolt = lookupBolt;
//	}

//	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
//		this._bolt.prepare(stormConf, context, collector);
//		this._watcher = Watcher.getWatcher(this.getClass());
//	}
//
//	public void execute(Tuple input) {
//		this._watcher.count();
//		this._bolt.execute(input);
//	}
//
//	public void cleanup() {
//		this._bolt.cleanup();
//	}
//
//	public void declareOutputFields(OutputFieldsDeclarer declarer) {
//		this._bolt.declareOutputFields(declarer);
//	}
//
//	public Map<String, Object> getComponentConfiguration() {
//		return this._bolt.getComponentConfiguration();
//	}

}
